feat: mic moc - #5486
Conversation
- Use RWMutex instead of Mutex for handler map access - Add atomic subscription counter to skip locking when no subscriptions exist - Hold read lock during handler iteration to prevent concurrent mutations - Extract payload once before dispatching to handlers - Inline getHandlers to reduce function call overhead - Apply optimizations consistently across gsoc, mic, and moc packages
sbackend123
left a comment
sbackend123
left a comment
There was a problem hiding this comment.
pkg/mic/mic.go and pkg/moc/moc.go are nearly identical — the only real difference is which field of the SOC is used as the subscription key (owner vs id).
I'd suggest extracting a shared listener implementation and keeping mic / moc as thin wrappers that only supply that key function. Both packages already expose a Listener interface; the shared code can implement it once, with a keyFn func(*soc.SOC) string (or []byte) passed in at construction time.
WDYT?
| defer l.handlersMu.Unlock() | ||
|
|
||
| l.handlers = make(map[string][]*Handler) // unset handlers on shutdown | ||
|
|
There was a problem hiding this comment.
possibly minor, but shouldn't we drop subCount to 0 here?
There was a problem hiding this comment.
In fact, is it not more go sytle to insert a context on the handler which is then calcelled when the listener is closed?
acud
left a comment
acud
left a comment
There was a problem hiding this comment.
Agree with @sbackend123 - I suggest to try to write this with generics.
|
Note on delivery semantics for subscription consumers: push-sync delivery is at-least-once — honest retries alone can deliver the same chunk repeatedly, and the handler dispatches to |
zelig
left a comment
zelig
left a comment
There was a problem hiding this comment.
I think this should be implemented with one generic soc hook , with argument which makes all parts of the soc available, not just the payload, ideally already parsed and validated, and could also tell which part the hook matched on.
See SWIP-60 also for topic match variants.
| defer l.handlersMu.Unlock() | ||
|
|
||
| l.handlers = make(map[string][]*Handler) // unset handlers on shutdown | ||
|
|
There was a problem hiding this comment.
In fact, is it not more go sytle to insert a context on the handler which is then calcelled when the listener is closed?
|
|
||
| // Handler defines code to be executed upon reception of a MIC message. | ||
| // It is used as a parameter definition. | ||
| type Handler func([]byte) |
There was a problem hiding this comment.
I think maybe this function should take context and a soc object. not sure why just the payload is in the argument. In all cases I have in mind, ID, owner, wrapped chunk etc are all relevant.
|
|
||
| for _, hh := range h { | ||
| go func(hh Handler) { | ||
| hh(payload) |
|
|
||
| // Handler defines code to be executed upon reception of a MOC message. | ||
| // It is used as a parameter definition. | ||
| type Handler func([]byte) |
There was a problem hiding this comment.
in fact if the handler is taking ctx , and a soc
4 participants
Description
Adds WebSocket subscription endpoints for receiving real-time Single Owner Chunk (SOC) payloads filtered by owner address (MIC) or identifier (MOC):
/mic/subscribe/{owner}/moc/subscribe/{id}Both listeners integrate with push/pull sync to dispatch matching SOC payloads to WebSocket subscribers.
Includes optimized SOC listener implementation using
RWMutexwith atomic fast-path to minimize lock contention on the hot sync path. The same optimization is applied to the existing GSOC listener for consistency.Open API Spec Version Changes (if applicable)
8.1.0 → 8.2.0 (new endpoints are additive features per SemVer)
Motivation and Context (Optional)
Enables real-time notifications for specific SOC patterns without polling.
Related Issue (Optional)
MIC/MOC SWIP
Checklist
Screenshots (if appropriate):
AI Disclosure